/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package elasticsearch

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"reflect"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/olivere/elastic/v7"
	"hundsun.com/hsl/hschain/protos"
)

var (
	errEmptyIndexName      = errors.New("indexName cannot be empty")
	errEmptySearchRequest  = errors.New("searchRequest cannot be empty")
	errNilDoc              = errors.New("document cannot be nil")
	errEmptyDocumentList   = errors.New("documents cannot be empty")
	errEmptyDocID          = errors.New("docID cannot be empty")
	errNilDocIDGenerator   = errors.New("docIDGenerator cannot be nil")
	ErrResultEmptyDocument = errors.New("document result is empty")
)

// DocIndexIdGenerator 为指定的ProtobufDocument类型对象生成index和docID
type DocIndexIdGenerator func(doc *protos.EnhancedProtoMessage) (index string, docID string)

// GetDocRequest 单个get文档请求
type GetDocRequest struct {
	Index string
	DocID string
}

// GetDocResult 单个get请求对应的查询结果
type GetDocResult struct {
	IndexName string
	DocID     string
	Found     bool
	Source    []byte
}

// BulkDocRequest bulk操作(statedbIndexBody/delete)时的单个请求
type BulkDocRequest interface {
	//GetXXXNamespace 获取请求操作的索引名称
	GetIndex() string

	//GetXXXKey 获取索引操作的文档ID
	GetDocID() string

	//IsDelete 是否是删除操作
	IsDelete() bool

	//GetDocBody 获取文档内容
	GetDocBody() map[string]interface{}
}

type bulkIndexDocRequest struct {
	doc       *protos.EnhancedProtoMessage
	generator DocIndexIdGenerator
}

func (request *bulkIndexDocRequest) GetIndex() string {
	var index string
	if request.generator != nil {
		index, _ = request.generator(request.doc)
	} else {
		index = (*request.doc).GetXXXNamespace()
	}
	return index
}

func (request *bulkIndexDocRequest) GetDocID() string {
	var docID string
	if request.generator != nil {
		_, docID = request.generator(request.doc)
	} else {
		docID = (*request.doc).GetXXXKey()
	}
	return docID
}

func (request *bulkIndexDocRequest) IsDelete() bool {
	return false
}

func (request *bulkIndexDocRequest) GetDocBody() map[string]interface{} {
	if docBody, err := convertProtobufDocumentToDocBody(request.doc); err != nil {
		return nil
	} else {
		return docBody
	}
}

// NewBulkIndexDocRequest 新建bulk操作中的单个index操作
func NewBulkIndexDocRequest(doc protos.EnhancedProtoMessage, generator DocIndexIdGenerator) *bulkIndexDocRequest {
	if doc == nil {
		return nil
	}
	return &bulkIndexDocRequest{doc: &doc, generator: generator}
}

// bulkDeleteDocRequest 删除文档请求
type bulkDeleteDocRequest struct {
	Index string
	DocID string
}

func (item *bulkDeleteDocRequest) GetIndex() string {
	return item.Index
}

func (item *bulkDeleteDocRequest) GetDocID() string {
	return item.DocID
}

func (item *bulkDeleteDocRequest) IsDelete() bool {
	return true
}

func (item *bulkDeleteDocRequest) GetDocBody() map[string]interface{} {
	return nil
}

// NewBulkDeleteDocRequest 新建bulk操作中的单个delete操作
func NewBulkDeleteDocRequest(index string, docID string) *bulkDeleteDocRequest {
	if index == "" {
		return nil
	}
	if docID == "" {
		return nil
	}
	return &bulkDeleteDocRequest{index, docID}
}

func validateProtobufDocument(doc *protos.EnhancedProtoMessage, generator DocIndexIdGenerator) (string, string, error) {
	if doc == nil {
		return "", "", errNilDoc
	}
	var index, docID string
	if generator != nil {
		index, docID = generator(doc)
	} else {
		index, docID = (*doc).GetXXXNamespace(), (*doc).GetXXXKey()
	}

	if docID == "" {
		return "", "", errEmptyDocID
	}
	if index == "" {
		return "", "", errEmptyIndexName
	}
	return index, docID, nil
}

func convertProtobufDocumentToDocBody(doc *protos.EnhancedProtoMessage) (map[string]interface{}, error) {
	if doc == nil {
		return nil, nil
	}
	docBody := make(map[string]interface{})
	if payload, err := proto.Marshal(*doc); err != nil {
		return nil, err
	} else {
		docBody[protos.EsDocFieldPayload] = payload
	}
	searchFields := (*doc).GetXXXSearchFields()
	for k, v := range searchFields {
		docBody[k] = v
	}
	return docBody, nil
}

func validateBulkRequest(req BulkDocRequest) (string, string, error) {
	if req == nil {
		return "", "", errors.New("req cannot be nil")
	}
	index, docID := req.GetIndex(), req.GetDocID()
	if docID == "" {
		return "", "", errEmptyDocID
	}
	if index == "" {
		return "", "", errEmptyIndexName
	}
	return index, docID, nil
}

// BulkResponseItem bulk操作时单个请求的应答
type BulkResponseItem struct {
	IndexName string
	DocID     string
	// status [200, 201, 404]
	Status int
}

// TermQuery 查询过滤器（And）
type TermQuery struct {
	Name  string
	Value string
}

// TermsQuery 查询过滤器（Or）
type TermsQuery struct {
	Name   string
	Values []string
}

// RangeQuery （字符串/数值类型）范围查询 [min, max)
type NumericRangeQuery struct {
	Field string
	Min   uint64
	Max   uint64
}

// Sort 排序
type Sort struct {
	Field     string
	Ascending bool
}

// SearchRequest 查询请求
type SearchRequest struct {
	Filters         []*TermQuery
	TermsCollection []*TermsQuery
	NumericRanges   []*NumericRangeQuery
	Sorts           []*Sort
	// 返回记录条数
	Size int
}

// SearchHit search时命中的一条记录
type SearchHit struct {
	IndexName string
	DocID     string
	Source    []byte
}

// SearchResult search返回结果
type SearchResult struct {
	Status int
	Total  int64
	Hits   []*SearchHit
}

// DeleteQueryRequest 查询删除请求
type DeleteQueryRequest struct {
	Filters []*TermQuery
}

// Document 代表ES中可以存储的文档
type Document interface {
	// GetID 获取文档ID
	GetID() string

	// GetIndexName 获取索引名称
	GetIndexName() string
}

// Client ES客户端，针对olivere/elastic/v7的封装
type Client struct {
	realClient *elastic.Client
}

// CreateIndex 创建索引（索引不存在才创建）。bool类型的返回值表示索引是否存在。
func (cli *Client) CreateIndex(indexName string, bodyString string) (bool, error) {
	if indexName == "" {
		return false, errEmptyIndexName
	}
	ctx := context.Background()
	if exists, err := cli.realClient.IndexExists(indexName).Do(ctx); err != nil {
		return false, err
	} else if !exists {
		if _, err := cli.realClient.CreateIndex(indexName).BodyString(bodyString).Do(ctx); err != nil {
			return false, err
		}
		return true, nil
	}
	return false, nil
}

// IndexDocumentWithRetry 将文档添加到指定索引中（支持重试）
func (cli *Client) IndexDocumentWithRetry(doc protos.EnhancedProtoMessage, generator DocIndexIdGenerator) error {
	index, docID, err := validateProtobufDocument(&doc, generator)
	if err != nil {
		return err
	}
	docBody, err := convertProtobufDocumentToDocBody(&doc)
	if err != nil {
		return err
	}

	defer log.ElapsedTime("IndexDocumentWithRetry", time.Now())

	var count = 0
	for {
		_, err = cli.realClient.Index().
			Index(index).
			Id(docID).
			BodyJson(docBody).
			Refresh("true").
			Do(context.Background())
		if err == nil {
			return nil
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("IndexDocumentWithRetry - failed to save with index:'%s', id:'%s', err:'%v'", index, docID, err)
			return err
		}
		// wait for retry
		log.Errorf("IndexDocumentWithRetry - wait %ds to retry save with index:'%s', id:'%s', err:%s", count, index, docID, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}
}

// GetDocumentWithRetry 根据docID获取文档，bool类型返回值表示是否找得到。（支持重试）
func (cli *Client) GetDocumentWithRetry(request GetDocRequest, result interface{}) (bool, error) {
	if request.Index == "" {
		return false, errEmptyIndexName
	}
	if request.DocID == "" {
		return false, errEmptyDocID
	}

	defer log.ElapsedTime("GetDocumentWithRetry", time.Now())

	var rs *elastic.GetResult
	var err error
	var count = 0
	for {
		rs, err = cli.realClient.Get().Index(request.Index).Id(request.DocID).Do(context.Background())
		if err == nil {
			break
		}
		if isNotFoundError(err) {
			return false, nil
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("GetDocumentWithRetry - failed to get with index:'%s', id:'%s', err:'%v'", request.Index, request.DocID, err)
			return false, err
		}
		// wait for retry
		log.Errorf("GetDocumentWithRetry - wait %ds to retry get with index:'%s', id:'%s', err:%s", count, request.Index, request.DocID, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}

	if len(rs.Source) > 0 {
		if err := protos.UnmarshalEnhancedProtoMessage(rs.Source, result); err != nil {
			return false, err
		} else {
			return true, nil
		}
	}
	return false, nil
}

// GetDocumentsWithRetry 根据docID批量获取文档（支持重试）
func (cli *Client) GetDocumentsWithRetry(multiGetItems []GetDocRequest, results []interface{}) ([]bool, error) {
	if len(multiGetItems) == 0 {
		return nil, errors.New("multiGetItems cannot be empty")
	}

	var rawItems []*elastic.MultiGetItem
	for _, item := range multiGetItems {
		if item.Index == "" {
			return nil, errEmptyIndexName
		}
		if item.DocID == "" {
			return nil, errEmptyDocID
		}

		rawItems = append(rawItems, elastic.NewMultiGetItem().Index(item.Index).Id(item.DocID))
	}

	defer log.ElapsedTime("GetDocumentsWithRetry", time.Now())

	var resp *elastic.MgetResponse
	var err error
	var count = 0
	for {
		resp, err = cli.realClient.Mget().Add(rawItems...).Do(context.Background())
		if err == nil {
			break
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("GetDocumentsWithRetry - failed to get , err:'%v'", err)
			return nil, err
		}
		// wait for retry
		log.Errorf("GetDocumentsWithRetry - wait %ds to retry get, err:%s", count, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}

	found := make([]bool, len(multiGetItems))
	for i, doc := range resp.Docs {
		found[i] = doc.Found
		if doc.Found {
			if err := protos.UnmarshalEnhancedProtoMessage(doc.Source, results[i]); err != nil {
				return found, err
			}
		}
	}
	return found, nil
}

func isNotFoundError(err error) bool {
	if esErr, ok := err.(*elastic.Error); ok && esErr.Status == http.StatusNotFound {
		return true
	}
	return false
}

// DeleteDocumentWithRetry 根据docID删除文档（支持重试）
func (cli *Client) DeleteDocumentWithRetry(indexName string, docID string) error {
	if indexName == "" {
		return errEmptyIndexName
	}
	if docID == "" {
		return errEmptyDocID
	}

	defer log.ElapsedTime("DeleteDocumentWithRetry", time.Now())

	var count = 0
	for {
		_, err := cli.realClient.Delete().
			Index(indexName).
			Id(docID).
			Refresh("true").
			Do(context.Background())
		if err == nil {
			return nil
		}
		if isNotFoundError(err) {
			return nil
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("DeleteDocumentWithRetry - failed to delete with index:'%s', id:'%s', err:'%v'", indexName, docID, err)
			return err
		}
		// wait for retry
		log.Errorf("DeleteDocumentWithRetry - wait %ds to retry delete with index:'%s', id:'%s', err:'%v'", indexName, docID, err)
		time.Sleep(time.Duration(count) * time.Second)
	}
}

// BulkIndexOrDeleteDocumentsWithRetry 对文档进行批量index、delete操作（支持重试）
func (cli *Client) BulkIndexOrDeleteDocumentsWithRetry(bulkRequestItems []BulkDocRequest) ([]*BulkResponseItem, error) {
	if len(bulkRequestItems) == 0 {
		return nil, errors.New("bulkRequestItems cannot be empty")
	}
	bulkReq := cli.realClient.Bulk().Refresh("true")
	for _, item := range bulkRequestItems {
		if item == nil {
			return nil, errors.New("nil found in bulkRequestItems")
		}
		index, docID, err := validateBulkRequest(item)
		if err != nil {
			return nil, err
		}
		if item.IsDelete() {
			bulkReq.Add(elastic.NewBulkDeleteRequest().Index(index).Id(docID))
		} else {
			bulkReq.Add(elastic.NewBulkIndexRequest().Index(index).Id(docID).Doc((item).GetDocBody()))
		}
	}

	defer log.ElapsedTime("BulkIndexOrDeleteDocumentsWithRetry", time.Now())

	var resp *elastic.BulkResponse
	var err error
	var count = 0
	for {
		resp, err = bulkReq.Do(context.Background())
		if err == nil {
			break
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("BulkIndexOrDeleteDocumentsWithRetry - failed to get or delete , err:'%v'", err)
			return nil, err
		}
		// wait for retry
		log.Errorf("BulkIndexOrDeleteDocumentsWithRetry - wait %ds to retry get or delete, err:%s", count, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}

	result := make([]*BulkResponseItem, len(resp.Items))
	for i, item := range resp.Items {
		for _, v := range item {
			result[i] = &BulkResponseItem{
				IndexName: v.Index,
				DocID:     v.Id,
				Status:    v.Status,
			}
		}
	}
	return result, nil
}

// SearchDocumentsWithRetry 查询指定索引中的文档（支持重试）（返回的是满足全部查询条件的结果）
func (cli *Client) SearchDocumentsWithRetry(indexName string, searchRequest *SearchRequest) (*SearchResult, error) {
	if indexName == "" {
		return nil, errEmptyIndexName
	}
	if searchRequest == nil {
		return nil, errEmptySearchRequest
	}

	defer log.ElapsedTime("SearchDocumentsWithRetry", time.Now())

	searchService := cli.realClient.Search(indexName)

	boolQuery := elastic.NewBoolQuery()
	// search condition: and
	for i, f := range searchRequest.Filters {
		if f.Name == "" {
			return nil, fmt.Errorf("name of filters[%d] is empty", i)
		}
		boolQuery.Filter(elastic.NewTermQuery(f.Name, f.Value))
	}

	// search condition: or
	for i, t := range searchRequest.TermsCollection {
		if t.Name == "" {
			return nil, fmt.Errorf("name of terms[%d] is empty", i)
		}
		values := toInterfaceSlice(t.Values)
		if len(values) < 1 {
			return nil, errors.New("failed to convert terms")
		}
		// Attention: ...
		boolQuery.Filter(elastic.NewTermsQuery(t.Name, values...))
	}

	// search condition: range
	for i, n := range searchRequest.NumericRanges {
		if len(n.Field) < 1 || n.Min >= n.Max {
			return nil, fmt.Errorf("NumericRanges[%d] has wrong args(field:%s min:%d max:%d)", i, n.Field, n.Min, n.Max)
		}
		query := elastic.NewRangeQuery(n.Field)
		query.Gte(n.Min)
		query.Lt(n.Max)
		boolQuery.Filter(query)
	}

	searchService.Query(boolQuery)

	for i, s := range searchRequest.Sorts {
		if s.Field == "" {
			return nil, fmt.Errorf("name of sorts[%d] is empty", i)
		}
		searchService.Sort(s.Field, s.Ascending)
	}

	if searchRequest.Size > 0 {
		searchService.Size(searchRequest.Size)
	} else {
		// TODO scroll游标方式可获取全部命中的结果（但需通过游标多次查询）
		// search命中后返回的最大条数
		searchService.Size(10000)
	}

	// search
	var rawResult *elastic.SearchResult
	var err error
	var count = 0
	for {
		rawResult, err = searchService.Do(context.Background())
		if err == nil {
			break
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("SearchDocumentsWithRetry - failed to search , err:'%v'", err)
			return nil, err
		}
		// wait for retry
		log.Errorf("SearchDocumentsWithRetry - wait %ds to retry search, err:%s", count, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}

	// result
	hits := make([]*SearchHit, len(rawResult.Hits.Hits))
	for i, h := range rawResult.Hits.Hits {
		hits[i] = &SearchHit{
			IndexName: h.Index,
			DocID:     h.Index,
			Source:    h.Source,
		}
	}
	result := &SearchResult{
		Status: rawResult.Status,
		Total:  rawResult.TotalHits(),
		Hits:   hits,
	}
	return result, nil
}

// DeleteByQueryWithRetry 通过查询删除（支持重试）
func (cli *Client) DeleteByQueryWithRetry(indexName string, deleteQueryRequest *DeleteQueryRequest) error {
	if indexName == "" {
		return errEmptyIndexName
	}
	if deleteQueryRequest == nil {
		return errEmptySearchRequest
	}

	searchService := cli.realClient.DeleteByQuery(indexName)
	if len(deleteQueryRequest.Filters) > 0 {
		boolQuery := elastic.NewBoolQuery()
		for i, f := range deleteQueryRequest.Filters {
			if f.Name == "" {
				return fmt.Errorf("name of filters[%d] is empty", i)
			}
			boolQuery.Filter(elastic.NewTermQuery(f.Name, f.Value))
		}
		searchService.Query(boolQuery)
	}

	defer log.ElapsedTime("DeleteByQueryWithRetry", time.Now())

	// delete
	var count = 0
	for {
		_, err := searchService.Do(context.Background())
		if err == nil {
			return nil
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("DeleteByQueryWithRetry - failed to delete , err:'%v'", err)
			return err
		}
		// wait for retry
		log.Errorf("DeleteByQueryWithRetry - wait %ds to retry delete, err:%s", count, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}
}

func toInterfaceSlice(slice interface{}) []interface{} {
	s := reflect.ValueOf(slice)
	if s.Kind() != reflect.Slice {
		return nil
	}

	ret := make([]interface{}, s.Len())

	for i := 0; i < s.Len(); i++ {
		ret[i] = s.Index(i).Interface()
	}

	return ret
}

// CountIndices 获取索引的文档条数；入参为多个索引时，返回则为其总和；（支持重试）
func (cli *Client) CountIndices(indices ...string) (int64, error) {
	if len(indices) < 1 {
		return -1, errEmptyIndexName
	}

	defer log.ElapsedTime("CountIndices", time.Now())

	var rs int64
	var err error
	var count = 0
	for {
		rs, err = cli.realClient.Count(indices...).Do(context.Background())
		if err == nil {
			return rs, nil
		}

		if isNotFoundError(err) {
			return -1, ErrResultEmptyDocument
		}

		count++
		// return with err
		if count > RetryLimit {
			log.Errorf("CountIndices - failed to count, err:'%v'", err)
			return -1, err
		}
		// wait for retry
		log.Errorf("CountIndices - wait %ds to retry count, err:%s", count, err.Error())
		time.Sleep(time.Duration(count) * time.Second)
	}
}
